﻿using System;
using System.Collections.Generic;
using System.Collections.Specialized;
using System.Configuration;
using System.Globalization;
using System.Text;

namespace CacheNaga
{
	/// <summary>
	/// Memcached client
	/// </summary>
	internal class MemcachedClient
	{
		#region Static fields & methods

		/// <summary>
		/// 所有memcached client实例缓存
		/// </summary>
		private static Dictionary<string, MemcachedClient> instances = new Dictionary<string, MemcachedClient>();
		private static LogAdapter logger = LogAdapter.GetLogger(typeof(MemcachedClient));

		/// <summary>
		/// 创建实例
		/// </summary>
		/// <param name="name">实例名</param>
		/// <param name="servers">使用逗号分隔的服务器ip地址,ip格式为ip:port,ip可以使用计算机名,如果port未设置默认使用11211端口</param>
		public static void Setup(string name, string[] servers)
		{
			if (instances.ContainsKey(name))
			{
				throw new ConfigurationErrorsException(name + " 已经存在");
			}
			instances[name] = new MemcachedClient(name, servers);
		}

		/// <summary>
		/// 检查给定的memcached client实例是否存在
		/// </summary>
		/// <param name="name">实例名</param>
		/// <returns></returns>
		public static bool Exists(string name)
		{
			return instances.ContainsKey(name);
		}

		/// <summary>
		/// 默认使用default
		/// </summary>
		private static MemcachedClient defaultInstance = null;
		public static MemcachedClient GetConfigInstance()
		{
			return defaultInstance ?? (defaultInstance = GetInstance("default"));
		}

		/// <summary>
		/// 根据配置名称获取MemcachedClient实例
		/// </summary>
		/// <param name="name">实例名</param>
		/// <returns><see cref="FwMemcachedClient"/>实例</returns>
		public static MemcachedClient GetInstance(string name)
		{
			MemcachedClient c;
			if (instances.TryGetValue(name, out c))
			{
				return c;
			}
			else
			{
				NameValueCollection config = ConfigurationManager.GetSection("cacheNaga") as NameValueCollection;
				if (config != null && !String.IsNullOrEmpty(config.Get(name)))
				{
					Setup(name, config.Get(name).Split(new char[] { ',' }));
					return GetInstance(name);
				}
				throw new ConfigurationErrorsException(name + "  -> 该实例不存在");
			}
		}

		#endregion

		#region Fields & Properties

		public readonly string Name;
		private readonly ServerPool serverPool;

		/// <summary>
		/// 设置缓存key的前缀(此前缀不参与consistant hashing计算)
		/// </summary>
		public string KeyPrefix { get; set; }

		/// <summary>
		/// 与memcached服务器的通讯超时时间,单位是毫秒,默认为2000
		/// </summary>
		public int SendReceiveTimeout { get { return serverPool.SendReceiveTimeout; } set { serverPool.SendReceiveTimeout = value; } }

		/// <summary>
		/// socket pool保存socket的最小数量.默认为5
		/// </summary>
		public uint MinPoolSize
		{
			get { return serverPool.MinPoolSize; }
			set
			{
				if (value > MaxPoolSize) { throw new ConfigurationErrorsException("MaxPoolSize:" + value + " 必须小于MaxPoolSize:" + MaxPoolSize); }
				serverPool.MinPoolSize = value;
			}
		}

		/// <summary>
		/// socket pool保存socket的最大数量.默认为10
		/// </summary>
		public uint MaxPoolSize
		{
			get { return serverPool.MaxPoolSize; }
			set
			{
				if (value < MinPoolSize) { throw new ConfigurationErrorsException("MaxPoolSize:" + value + " 必须大于MinPoolSize: " + MinPoolSize); }
				serverPool.MaxPoolSize = value;
			}
		}

		/// <summary>
		/// Socket回收时间,默认为30分钟
		/// </summary>
		public TimeSpan SocketRecycleAge { get { return serverPool.SocketRecycleAge; } set { serverPool.SocketRecycleAge = value; } }

		private uint compressionThreshold = 1024 * 32; //32kb

		/// <summary>
		/// strings, byte arrays和object类型的对象在超过一定大小时启用压缩的临界值,
		/// 默认为1024*64字节(64kb)
		/// </summary>
		public uint CompressionThreshold { get { return compressionThreshold; } set { compressionThreshold = value; } }

		#endregion

		#region Private .ctor

		private MemcachedClient(string name, string[] hosts)
		{
			if (String.IsNullOrEmpty(name))
			{
				throw new ConfigurationErrorsException("MemcachedClient实例不能为空");
			}
			if (hosts == null || hosts.Length == 0)
			{
				throw new ConfigurationErrorsException("必须指定服务器列表");
			}

			Name = name;
			serverPool = new ServerPool(hosts);
		}

		#endregion

		/* Memcached Protocol中的术语解释
		 * <bytes>是之后返回的数据块的大小(不包含\r\n)
		 * <cas unique>是标识该项的唯一的64位整数
		 * <data block>是该项对应的值
		 * <time>时间有两种,一种是针对当前时间再增加的秒数,一种是unix时间格式(GMT).使用秒数计算最大只能到30天,超过这个值会被认为是unix time来计算
		 */

		//---memcached操作开始

		#region 缓存功能方法

		#region [Set] 添加缓存,已存在的缓存会被覆盖

		public bool Set(string key, object value) { return ExecuteStoreCommandAdapter("set", key, true, value, Helper.Hash(key), 0); }
		public bool Set(string key, object value, TimeSpan expiry) { return ExecuteStoreCommandAdapter("set", key, true, value, Helper.Hash(key), (int)expiry.TotalSeconds); }
		public bool Set(string key, object value, DateTime expiry) { return ExecuteStoreCommandAdapter("set", key, true, value, Helper.Hash(key), Helper.GetUnixTime(expiry)); }

		#endregion

		#region [Add] 如果key不存在则添加到缓存

		public bool Add(string key, object value) { return ExecuteStoreCommandAdapter("add", key, true, value, Helper.Hash(key), 0); }

		public bool Add(string key, object value, DateTime expiry) { return ExecuteStoreCommandAdapter("add", key, true, value, Helper.Hash(key), Helper.GetUnixTime(expiry)); }

		public bool Add(string key, object value, TimeSpan expiry) { return ExecuteStoreCommandAdapter("add", key, true, value, Helper.Hash(key), (int)expiry.TotalSeconds); }

		#endregion

		#region [Replace] 如果key存在的话替换它的值

		public bool Replace(string key, object value) { return ExecuteStoreCommandAdapter("replace", key, true, value, Helper.Hash(key), 0); }
		public bool Replace(string key, object value, TimeSpan expiry) { return ExecuteStoreCommandAdapter("replace", key, true, value, Helper.Hash(key), (int)expiry.TotalSeconds); }
		public bool Replace(string key, object value, DateTime expiry) { return ExecuteStoreCommandAdapter("replace", key, true, value, Helper.Hash(key), Helper.GetUnixTime(expiry)); }

		#endregion

		#region [Append] 如果key存在将值添加到key对应的值上(不更改过期时间)

		public bool Append(string key, object value) { return ExecuteStoreCommandAdapter("append", key, true, value, Helper.Hash(key)); }

		#endregion

		#region [Prepend] 如果key存在将值添加到key的前面(不更改过期时间)

		public bool Prepend(string key, object value) { return ExecuteStoreCommandAdapter("prepend", key, true, value, Helper.Hash(key)); }

		#endregion

		#region [Delete] 删除缓存

		public bool Delete(string key) { return ExecuteDeleteCommand(key, true, Helper.Hash(key), 0); }
		public bool Delete(string key, TimeSpan delay) { return ExecuteDeleteCommand(key, true, Helper.Hash(key), (int)delay.TotalSeconds); }
		public bool Delete(string key, DateTime delay) { return ExecuteDeleteCommand(key, true, Helper.Hash(key), Helper.GetUnixTime(delay)); }

		#endregion

		#region [SetCounter] 设置计数器

		public bool SetCounter(string key, ulong value) { return Set(key, value.ToString(CultureInfo.InvariantCulture)); }
		public bool SetCounter(string key, ulong value, TimeSpan expiry) { return Set(key, value.ToString(CultureInfo.InvariantCulture), expiry); }
		public bool SetCounter(string key, ulong value, DateTime expiry) { return Set(key, value.ToString(CultureInfo.InvariantCulture), expiry); }

		#endregion

		#region [GetCounter] 读取计数器

		public ulong? GetCounter(string key) { return ExecuteGetCounterCommandAdapter(key, true, Helper.Hash(key)); }
		public ulong?[] GetCounter(string[] keys) { return ExecuteGetCounterCommandAdapter(keys, true, Helper.Hash(keys)); }

		#endregion

		#region [Increment] 计数器递增

		/// <summary>
		/// 执行incr命令
		/// </summary>
		public ulong? Increment(string key, ulong value) { return ExecuteIncrDecrCommand("incr", key, true, value, Helper.Hash(key)); }

		#endregion

		#region [Decrement] 计数器递减少
		/// <summary>
		/// 执行decr命令
		/// </summary>
		/// <remarks>
		/// 如果decr之后的值小于0结果将为0
		/// </remarks>
		public ulong? Decrement(string key, ulong value) { return ExecuteIncrDecrCommand("decr", key, true, value, Helper.Hash(key)); }

		#endregion

		#region [Get] 获取缓存数据

		/// <summary>
		/// 根据Key获取缓存的内容
		/// 如果不存在返回null;如果传递key的数组将返回相同大小的包含返回值的object数组
		/// </summary>
		public object Get(string key) { return ExecuteGetCommand("get", key, true, Helper.Hash(key)); }

		/// <summary>
		/// 根据key批量获取缓存
		/// </summary>
		/// <param name="keys"></param>
		/// <returns></returns>
		public object[] Get(string[] keys) { return ExecuteGetCommand("get", keys, true, Helper.Hash(keys)); }

		#endregion

		#endregion

		#region 执行Memcached命令的adapters

		/// <summary>
		/// [Set, Add, Replace] command adapter
		/// </summary>
		private bool ExecuteStoreCommandAdapter(string command, string key, bool keyIsChecked, object value, uint hash, int expiry)
		{
			return ExecuteStoreCommand(command, key, keyIsChecked, value, hash, expiry, 0).StartsWith("STORED");
		}

		/// <summary>
		/// [Append, Prepend] command adapter
		/// </summary>
		private bool ExecuteStoreCommandAdapter(string command, string key, bool keyIsChecked, object value, uint hash)
		{
			return ExecuteStoreCommand(command, key, keyIsChecked, value, hash, 0, 0).StartsWith("STORED");
		}

		private ulong? ExecuteGetCounterCommandAdapter(string key, bool keyIsChecked, uint hash)
		{
			ulong parsedLong;
			return ulong.TryParse(ExecuteGetCommandWithUnique("get", key, keyIsChecked, hash) as string, out parsedLong) ? (ulong?)parsedLong : null;
		}

		private ulong?[] ExecuteGetCounterCommandAdapter(string[] keys, bool keysAreChecked, uint[] hashes)
		{
			ulong?[] results = new ulong?[keys.Length];
			object[] values = ExecuteGetCommand("get", keys, keysAreChecked, hashes);
			for (int i = 0; i < values.Length; i++)
			{
				ulong parsedLong;
				results[i] = ulong.TryParse(values[i] as string, out parsedLong) ? (ulong?)parsedLong : null;
			}
			return results;
		}

		#endregion

		#region 执行Socket请求的方法

		#region Set, Add, Replace, Append, Prepend对应的Socket请求

		/// <summary>
		/// 执行Add, Set, Replace, Append, Prepend相关的Socket请求,并将memcached给予的结果返回
		/// </summary>
		/// <remarks>
		/// 执行Memcached的Set, Add, Replace命令
		/// [命令格式]
		/// {command name} {key} {flags} {exptime} {bytes} [noreply]\r\n
		/// cas {key} {flags} {exptime} {bytes} {cas unqiue} [noreply]\r\n
		/// [返回格式]
		/// STORED\r\n 表示成功
		/// NOT_STORED\r\n 因为一些原因没有存储成功(不是异常)
		/// EXISTS\r\n 使用cas命令表示自从上次获取后已经被修改
		/// NOT_FOUND\r\n 使用cas命令时该key不存在或已经被删除
		/// 
		/// [注意]
		/// set:无条件存储
		/// add:当该key不存在时才存储,如果已存在返回NOT_STORED
		/// replace:仅当该key存在时存储,如果不存在返回NOT_STORED
		/// 
		/// {flags}是32-bit unsigned int(1.2.1+version,之前是16bit的) 存储附属信息 在该memcached client里用于存储类型.
		/// {exptime}等于上文{time}的注释
		/// </remarks>
		private string ExecuteStoreCommand(string command, string key, bool keyIsChecked, object value, uint hash, int expiry, ulong unique)
		{
			if (!keyIsChecked)
			{
				Helper.ValidateKey(key);
			}

			return serverPool.Execute<string>(hash, "", delegate(PooledSocket socket)
			{
				SerializedType type;
				byte[] bytes;

				try
				{
					bytes = Serializer.Serialize(value, out type, CompressionThreshold);
				}
				catch (Exception e)
				{
					//序列化失败返回string.Empty
					logger.Error("Error serializing object for key '" + key + "'.", e);
					return "";
				}

				//创建commandline
				string commandline = "";
				switch (command)
				{
					case "set":
					case "add":
					case "replace":
						commandline = command + " " + KeyPrefix + key + " " + (ushort)type + " " + expiry + " " + bytes.Length + "\r\n";
						break;
					case "append":
					case "prepend":
						commandline = command + " " + KeyPrefix + key + " 0 0 " + bytes.Length + "\r\n";
						break;
				}

				//发送commandline
				socket.Write(commandline);
				socket.Write(bytes);
				socket.Write("\r\n");
				return socket.ReadResponse();
			});
		}

		#endregion

		#region Delete对应的Socket请求

		/*
		 * 对应Memcached的delete方法
		 * 
		 * [命令格式]
		 * delete <key> [<time>] [noreply]\r\n
		 * <time>表示让Memcached将该对象放入删除队列,并在n秒或者某个unix time时将其删除,默认为0.
		 * noreply表示让Memcached不返回任何结果
		 * 
		 * [返回格式]
		 * DELETED\r\n 表示成功
		 * NOT_FOUND\r\n 表示该项未找到
		 */
		private bool ExecuteDeleteCommand(string key, bool keyIsChecked, uint hash, int time)
		{
			if (!keyIsChecked)
			{
				Helper.ValidateKey(key);
			}

			return serverPool.Execute<bool>(hash, false, delegate(PooledSocket socket)
			{
				string commandline;
				if (time == 0)
				{
					commandline = "delete " + KeyPrefix + key + "\r\n";
				}
				else
				{
					commandline = "delete " + KeyPrefix + key + " " + time + "\r\n";
				}
				socket.Write(commandline);
				return socket.ReadResponse().StartsWith("DELETED");
			});
		}

		#endregion

		#region Incr, Decr对应的Socket请求

		/*
		 * 执行Memcached的incr和decr命令
		 * 
		 * [命令格式]
		 * incr <key> <value> [noreply]\r\n
		 * decr <key> <value> [noreply]\r\n
		 * 
		 * [返回格式]
		 * NOT_FOUND\r\n 表示该key不存在
		 * <value>\r\表示新的值
		 * 
		 * 需要注意的是key是必须存在的,如果不存在直接报告NOT_FOUND
		 * value是64-bit整数,incr的时候越界会报告错误
		 * decr后的value如果小于0则结果为0
		 */
		/// <summary>
		/// 执行Incr/Decr方法
		/// </summary>
		/// <returns>执行结果后返回的值</returns>
		/// <remarks>
		/// 这里没有支持noreply特性,所以每次操作值会回显
		/// </remarks>
		private ulong? ExecuteIncrDecrCommand(string cmd, string key, bool keyIsChecked, ulong value, uint hash)
		{
			if (!keyIsChecked)
			{
				Helper.ValidateKey(key);
			}
			return serverPool.Execute<ulong?>(hash, null, delegate(PooledSocket socket)
			{
				string command = cmd + " " + KeyPrefix + key + " " + value + "\r\n";
				socket.Write(command);
				string response = socket.ReadResponse();
				//如果key不存在将会提示错误
				if (response.StartsWith("NOT_FOUND"))
				{
					return null;
				}
				else
				{
					return Convert.ToUInt64(response.TrimEnd('\0', '\r', '\n'));
				}
			});
		}

		#endregion

		#region Get对应的Socket请求
		/*
		 * 对应Memcached的get方法
		 * [命令格式]
		 * get <key>*\r\n
		 * gets <key>*\r\n
		 * 
		 * [返回格式]
		 * VALUE <key> <flags> <bytes> [<cas unique>]\r\n
		 * <data block>\r\n
		 */
		private object ExecuteGetCommand(string command, string key, bool keyIsChecked, uint hash)
		{
			if (!keyIsChecked)
			{
				Helper.ValidateKey(key);
			}

			object value = serverPool.Execute<object>(hash, null, delegate(PooledSocket socket)
			{
				socket.Write("get " + KeyPrefix + key + "\r\n");
				object _value;
				if (ReadValueForGetCommand(socket, out _value, out key))
				{
					socket.ReadLine();
				}
				return _value;
			});
			return value;
		}

		private object ExecuteGetCommandWithUnique(string command, string key, bool keyIsChecked, uint hash)
		{
			if (!keyIsChecked)
			{
				Helper.ValidateKey(key);
			}

			object value = serverPool.Execute<object>(hash, null, delegate(PooledSocket socket)
			{
				socket.Write(command + " " + KeyPrefix + key + "\r\n");
				object _value;
				if (ReadValue(socket, out _value, out key))
				{
					socket.ReadLine(); //Read the trailing END.
				}
				return _value;
			});
			return value;
		}

		/// <summary>
		/// 获取多个缓存
		/// </summary>
		/// <param name="command"></param>
		/// <param name="keys">key的集合</param>
		/// <param name="keysAreChecked"></param>
		/// <param name="hashes"></param>
		/// <returns></returns>
		private object[] ExecuteGetCommand(string command, string[] keys, bool keysAreChecked, uint[] hashes)
		{
			if (keys == null || hashes == null)
			{
				throw new ArgumentException("Keys and hashes arrays must not be null.");
			}
			if (keys.Length != hashes.Length)
			{
				throw new ArgumentException("Keys and hashes arrays must be of the same length.");
			}

			//如果只有一个key则直接从一台机器上返回结果
			if (keys.Length == 1)
			{
				return new object[] { ExecuteGetCommandWithUnique(command, keys[0], keysAreChecked, hashes[0]) };
			}

			//检查key
			if (!keysAreChecked)
			{
				for (int i = 0; i < keys.Length; i++)
				{
					Helper.ValidateKey(keys[i]);
				}
			}

			//根据hash分组到相同serverPool的key
			Dictionary<SocketPool, Dictionary<string, List<int>>> dict = new Dictionary<SocketPool, Dictionary<string, List<int>>>();
			for (int i = 0; i < keys.Length; i++)
			{
				Dictionary<string, List<int>> getsForServer;
				//计算hash到哪台memcached server上
				SocketPool pool = serverPool.GetSocketPool(hashes[i]);
				//如果dict中没有这个pool,则添加
				if (!dict.TryGetValue(pool, out getsForServer))
				{
					dict[pool] = getsForServer = new Dictionary<string, List<int>>();
				}
				//保证获取到的值按照传入的key的顺序
				List<int> positions;
				if (!getsForServer.TryGetValue(keys[i], out positions))
				{
					//这里的keys[i]是没有加前缀的,而后期从缓存中获取到的值要以添加了前缀的key
					//作为key来赋值,导致批量get方法获取到的值都是null.
					getsForServer[KeyPrefix + keys[i]] = positions = new List<int>();
				}
				positions.Add(i);
			}

			//保存返回的值
			object[] returnValues = new object[keys.Length];
			foreach (KeyValuePair<SocketPool, Dictionary<string, List<int>>> kv in dict)
			{
				serverPool.Execute(kv.Key, delegate(PooledSocket socket)
				{
					//构造Get key1 key2 key3格式
					StringBuilder getRequest = new StringBuilder(command);
					foreach (KeyValuePair<string, List<int>> key in kv.Value)
					{
						getRequest.Append(" ");
						//key是直接在dict里存在的
						//因此这里去掉KeyPrefix
						//getRequest.Append(KeyPrefix);
						getRequest.Append(key.Key);
					}
					getRequest.Append("\r\n");

					socket.Write(getRequest.ToString());

					object gottenObject;
					string gottenKey;
					while (ReadValue(socket, out gottenObject, out gottenKey))
					{
						foreach (int position in kv.Value[gottenKey])
						{
							returnValues[position] = gottenObject;
						}
					}
				});
			}
			return returnValues;
		}


		#endregion

		#endregion

		//---memcached操作结束

		#region Get的帮助方法

		/// <summary>
		/// 读取get命令的结果
		/// </summary>
		/// <remarks>
		/// 如果读取的项不存在或读取过程中出错返回null
		/// </remarks>
		private bool ReadValueForGetCommand(PooledSocket socket, out object value, out string key)
		{
			string response = socket.ReadResponse();
			string[] parts = response.Split(' '); //返回格式VALUE <key> <flags> <bytes> <cas unique>
			if (parts[0] == "VALUE")
			{
				key = parts[1];
				//从<flags>中读取该值对应的.net类型
				SerializedType type = (SerializedType)Enum.Parse(typeof(SerializedType), parts[2]);
				//根据<bytes>指名的值的大小创建byte数组
				byte[] bytes = new byte[Convert.ToUInt32(parts[3], CultureInfo.InvariantCulture)];

				socket.Read(bytes);
				socket.SkipUntilEndOfLine();

				//根据类型反序列化
				try
				{
					value = Serializer.DeSerialize(bytes, type);
				}
				catch (Exception e)
				{
					value = null;
					logger.Error("Error deserializing object for key '" + key + "' of type " + type + ".", e);
				}
				return true;
			}
			else
			{
				key = null;
				value = null;
				return false;
			}
		}

		private bool ReadValue(PooledSocket socket, out object value, out string key)
		{
			string response = socket.ReadResponse();
			string[] parts = response.Split(' '); //Result line from server: "VALUE <key> <flags> <bytes> <cas unique>"
			if (parts[0] == "VALUE")
			{
				key = parts[1];
				SerializedType type = (SerializedType)Enum.Parse(typeof(SerializedType), parts[2]);
				byte[] bytes = new byte[Convert.ToUInt32(parts[3], CultureInfo.InvariantCulture)];
				socket.Read(bytes);
				socket.SkipUntilEndOfLine(); //Skip the trailing \r\n
				try
				{
					value = Serializer.DeSerialize(bytes, type);
				}
				catch (Exception e)
				{
					//反序列化失败操作
					value = null;
					logger.Error("Error deserializing object for key '" + key + "' of type " + type + ".", e);
				}
				return true;
			}
			else
			{
				key = null;
				value = null;
				return false;
			}
		}

		#endregion
	}
}
